package com.lfs.market.canal;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.lfs.market.model.entity.User;
import com.lfs.market.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class CanalClientRunner implements CommandLineRunner {
    @Autowired
    private UserService userService;

    @Override
    public void run(String... args) throws Exception {
        String host = "192.168.187.135";
        int port = 11111;
        String destination = "example";
        // 账号密码不写也能运行
        String username = "canal_username";
        String password = "canal_password";
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(host, port), destination, username, password);

        // 每次获取的所有更新数据条数
        int batchSize = 100;

        while (true) {
            try {
                connector.connect();
                connector.subscribe("remote\\.remote_user");
                connector.rollback();

                log.info("Canal 客户端启动成功，开始监听...");

                while (true) {
                    Message message = connector.getWithoutAck(batchSize);
                    long batchId = message.getId();

                    if (batchId == -1 || message.getEntries().isEmpty()) {
                        Thread.sleep(3000);
                        continue;
                    }

                    for (CanalEntry.Entry entry : message.getEntries()) {
                        processEntry(entry);
                    }

                    connector.ack(batchId);
                }
            } catch (Exception e) {
                log.error("Canal 客户端异常", e);
                Thread.sleep(5000);
            } finally {
                connector.disconnect();
            }
        }
    }

    private void processEntry(CanalEntry.Entry entry) throws Exception {
        if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) return;

        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        CanalEntry.EventType eventType = rowChange.getEventType();
        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();

        // 每次增量的数据量并不大时直接保存
        for (CanalEntry.RowData rowData : rowDataList) {
            Map<String, Object> dataMap = new HashMap<>();
            if (eventType == CanalEntry.EventType.INSERT) {
                rowData.getAfterColumnsList().forEach(col -> dataMap.put(col.getName(), col.getValue()));
                User user = mapToEntity(dataMap, User.class);
                userService.save(user);
                log.info("插入数据: {}", user);
            } else if (eventType == CanalEntry.EventType.UPDATE) {
                // 更新时取 afterColumns
                rowData.getAfterColumnsList().forEach(col -> dataMap.put(col.getName(), col.getValue()));
                User user = mapToEntity(dataMap, User.class);
                userService.updateById(user);
                log.info("更新数据: {}", user);
            } else if (eventType == CanalEntry.EventType.DELETE) {
                rowData.getBeforeColumnsList().forEach(col -> dataMap.put(col.getName(), col.getValue()));
                User user = mapToEntity(dataMap, User.class);
                userService.removeById(user.getId());
                log.info("删除数据: {}", user.getId());
            }
        }
    }

    // 简单的 Map → Entity 映射（可用 BeanUtil 工具类替代）
    private <T> T mapToEntity(Map<String, Object> map, Class<T> entityClass) {
        return BeanUtil.mapToBean(map, entityClass, false, null);
    }
}